package com.ruirui.sinks;

import reactor.core.publisher.Sinks;

import java.util.concurrent.LinkedBlockingDeque;

public class TestSinks {

    public static void main(String[] args) throws Exception{
//        // 发送多个flux
//        Sinks.many();
//        //发送mono
//        Sinks.one();
        Sinks.many().unicast();//单模 只绑定单个订阅者 多个报错
        Sinks.many().multicast();//多模 绑定多个订阅者
        Sinks.many().replay();//重放 管道可以重放元素
        Sinks.Many<Object> many = Sinks.many().multicast()
                .onBackpressureBuffer( );
        //后台发送数据
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                many.tryEmitNext(i);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        //订阅
        many.asFlux().subscribe(System.out::println);
        many.asFlux().subscribe(e-> System.out.println("e2:" + e));
        System.in.read();
    }
}
